package com.cn.kafka.producer;

import java.util.Date;
import java.util.UUID;

import com.alibaba.fastjson.JSON;
import com.cn.kafka.enrity.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;


/**
 * 发送消息的 生产者
 */
@Component
public class Sender {

    /**
     * 在其他地方直接注入使用也行
     */
    @Autowired
    private KafkaTemplate kafkaTemplate;


    public void sendMessage() {
        Message m = new Message();
        m.setId(System.currentTimeMillis());
        m.setMsg(UUID.randomUUID().toString());
        m.setSendTime(new Date());
        kafkaTemplate.send("test1", JSON.toJSONString(m));

        kafkaTemplate.execute(producer -> {
            return null;
        });

        /*kafkaTemplate.execute(new KafkaOperations.ProducerCallback<String, String, Object>() {
            @Override
            public Object doInKafka(Producer<String, String> producer) {
                //这里可以编写kafka原生的api操作
                return null;
            }
        });*/
    }

    public void sendMessage2() {
        Message m = new Message();
        m.setId(System.currentTimeMillis());
        m.setMsg(UUID.randomUUID().toString());
        m.setSendTime(new Date());
        kafkaTemplate.send("test2", JSON.toJSONString(m));
    }
}

